package com.atguigu.flink.datastream.transform;

import com.atguigu.flink.func.ClickSource;
import com.atguigu.flink.pojo.Event;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SideOutputDataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

/**
 * @author WEIYUNHUI
 * @date 2023/6/14 9:13
 *
 * 分流：
 *   1. 简单实现 ， 通过filter算子来实现。
 *   2. 使用侧输出流实现
 *
 */
public class Flink08_SplitStream {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStreamSource<Event> ds = env.addSource(new ClickSource());
        //将不同用户的数据分到不同的流中
        /*
        //简单实现 : filter
        SingleOutputStreamOperator<Event> zhangsDs = ds.filter(event -> "Zhangs".equals(event.getUser()));
        SingleOutputStreamOperator<Event> LisiDs = ds.filter(event -> "Lisi".equals(event.getUser()));
        SingleOutputStreamOperator<Event> tomDs = ds.filter(event -> "Tom".equals(event.getUser()));
        SingleOutputStreamOperator<Event> jerryDs = ds.filter(event -> "Jerry".equals(event.getUser()));
        zhangsDs.print("zhangs");
        LisiDs.print("Lisi");
        tomDs.print("tom");
        jerryDs.print("jerry");

         */

        //使用侧输出流实现:

        //创建输出标签
        //OutputTag<Event> zhangsOutputTag = new OutputTag<Event>("zhangs"){};
        OutputTag<Event> zhangsOutputTag = new OutputTag<>("zhangs" , Types.POJO(Event.class));
        OutputTag<Event> lisiOutputTag = new OutputTag<Event>("lisi"){};
        OutputTag<Event> tomOutputTag = new OutputTag<Event>("tom"){};

        SingleOutputStreamOperator<String> mainDs = ds.process(
                new ProcessFunction<Event, String>() {
                    @Override
                    public void processElement(Event value, ProcessFunction<Event, String>.Context ctx, Collector<String> out) throws Exception {
                        if ("Zhangs".equals(value.getUser())) {
                            //放入Zhangs流
                            ctx.output(zhangsOutputTag, value);
                        } else if ("Lisi".equals(value.getUser())) {
                            //放入Lisi流
                            ctx.output(lisiOutputTag, value);
                        } else if ("Tom".equals(value.getUser())) {
                            //放入Tom流
                            ctx.output(tomOutputTag, value);
                        } else {
                            //主流 Jerry Peiqi
                            out.collect(value.toString());
                        }
                    }
                }
        );

        mainDs.print("main");

        //通过主流获取侧流
        mainDs.getSideOutput(zhangsOutputTag).print("zhangs");
        mainDs.getSideOutput(lisiOutputTag).print("lisi");
        mainDs.getSideOutput(tomOutputTag).print("tom");


        try {
            env.execute();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
}
